package com.sunday.common.reactor.e2_Schedulers;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.Disposable;
import reactor.core.Exceptions;
import reactor.core.publisher.*;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import reactor.util.retry.Retry;

import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;

/**
 * https://projectreactor.io/docs/core/release/reference/#schedulers
 */
@Slf4j
public class ThreadingAndSchedulers {

    /**
     * subscribeOn 和 publishOn 是 Reactor（以及类似的响应式编程库）中用于控制流在不同线程上执行的操作符。
     * subscribeOn：用于指定数据流的订阅过程所发生的线程。它会影响订阅代码以及向上游请求数据的操作所发生的线程。只有第一次调用 subscribeOn 会生效，后续的 subscribeOn 调用将被忽略。
     * publishOn：用于指定数据流中后续操作所发生的线程。它会影响数据处理和触发下游操作（例如 map、filter 等）所发生的线程。
     * 举例来说，如果你在一个链式操作中使用了 subscribeOn，那么这个指定的线程就会影响整个数据流的订阅过程。而如果你使用了 publishOn，那么只有后续的操作会在指定的线程上执行。
     * 总之，subscribeOn 是用于指定整个订阅过程所发生的线程，而 publishOn 用于指定数据处理操作所发生的线程。
     */

    /**
     * 4.5. 线程和调度器
     * Reactor 和 RxJava 一样，可以被认为是与并发无关的。也就是说，它没有 强制实施并发模型。相反，它让你，开发人员，在指挥。然而 这并不妨碍库帮助您实现并发性。
     */
    @Test
    public void schedulers() throws InterruptedException {

        // 组装在线程中。Mono<String>main
        final Mono<String> mono = Mono.just("hello ");
        Thread t = new Thread(() -> mono
                .map(msg -> msg + "thread ")
                // 	但是，它是在线程中订阅的。Thread-0
                .subscribe(v ->
                        // 	因此，实际上 和 callback 都在map onNextThread-0
                        log.info("{} -> {}", v, Thread.currentThread().getName())
                )
        );
        t.start();
        t.join();
    }

    /**
     * Schedulers 类具有静态方法，这些方法允许访问以下执行上下文：
     * 无执行上下文（）：在处理时，提交的会直接执行，有效地在当前（可以看作是“空对象”或空操作）上运行它们。Schedulers.immediate() RunnableThreadScheduler
     * 单个可重用线程 （）。请注意，此方法重用 所有调用方的同一线程，直到调度程序被释放。如果您想要每次通话 专用线程，用于每个调用。Schedulers.single() Schedulers.newSingle()
     * 无界弹性线程池 （）。这个不再是首选 随着 的引入，因为它有隐藏背压的倾向 问题并导致线程过多（见下文）。Schedulers.elastic() Schedulers.boundedElastic()
     * 有界弹性线程池 （）。像它的前身一样，它 根据需要创建新的工作线程池，并重用闲置的工作线程池。闲置时间过长（默认值为 60 秒）的工作线程池是 也被处理掉了。与前身不同，它对可以创建的后备线程数有上限（默认值为 CPU 内核数 x 10）。 在达到上限后提交的多达 100 000 个任务将排队，并将在线程可用时重新安排 （当有延迟进行调度时，延迟在线程可用时开始）。这是 I/O 阻塞工作的更好选择。 是一种方便的方法，可以为阻塞进程提供自己的线程，以便 它不会占用其他资源。请参阅How Do I Wrap a Synchronous， Blocking Call？，但不会用新线程给系统带来太大压力。Schedulers.boundedElastic() elastic() elastic() Schedulers.boundedElastic()
     * 为并行工作 （） 进行优化的固定工作线程池。它 创建与拥有的 CPU 内核一样多的工作线程。Schedulers.parallel()
     */

    /**
     * 4.5.1. 方法publishOn
     * publishOn以与任何其他运算符相同的方式应用，在中间 订阅者链。它从上行获取信号并在下游重放，而 对来自关联 .因此，它会影响后续运算符的执行位置（直到另一个运算符 链接），如下所示：SchedulerpublishOn
     * 将执行上下文更改为由ThreadScheduler
     * 根据规范，调用是按顺序进行的，因此这会占用单个线程onNext
     * 除非它们在特定的 ，运算符上工作，否则在同一线程上继续执行 Scheduler publishOn
     */
    @Test
    public void publishOn() throws InterruptedException {

        //  <1> 创建一个由四个实例支持的新实例。SchedulerThread
        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 2)
                // doOnNext 作用是在每次 onNext 之前执行一段逻辑。
                .doOnNext(i -> log.info(i + ""))
                // 第一个在 <5> 中的匿名线程上运行。map Thread-0
                .map(i -> 10 + i)
                // 在从 <1> 拾取的序列上切换整个序列。publishOn Thread
                .publishOn(s)
                // 第二个运行在 <1> 上。mapThread
                .map(i -> "value " + i);

        //  <5> 这个匿名是发生订阅的匿名者。 打印发生在最新的执行上下文上，该上下文来自 。Thread publishOn parallel-scheduler-1
        new Thread(() -> flux.subscribe(i -> log.info(i))).start();

        TimeUnit.SECONDS.sleep(2);
    }

    /**
     * 4.5.2. 方法subscribeOn
     * subscribeOn适用于订阅过程，当后向链是 构建。通常建议将其紧接放在数据源之后， 作为中间运算符，可以影响执行的上下文。
     * 但是，这并不影响 后续调用的行为 — 它们仍会切换 它们之后的链条部分。publishOn
     * 更改整个运营商链的订阅方式Thread
     * 从Scheduler
     * 仅有效调用下游链中最接近的调用 将订阅和请求信号安排给可以的源或运营商 拦截它们 （， ）。使用多个调用将 引入没有价值的不必要的线程开关。subscribeOndoFirstdoOnRequestsubscribeOn
     */
    @Test
    public void subscribeOn() throws InterruptedException {
        //  <1> 创建一个由四个实例支持的新实例。SchedulerThread
        Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 2)
                .doOnNext(i -> log.info(i + ""))
                // 第一个运行在这四个线程之一上......map
                .map(i -> 10 + i)
                // 因为从订阅时间 （<5>） 切换整个序列。subscribeOn
                .subscribeOn(s)
                // 第二个也在同一线程上运行。map
                .map(i -> "value " + i);

        //  <5> 这个匿名是发生订阅的匿名者。 打印发生在最新的执行上下文上，该上下文来自 。Thread publishOn parallel-scheduler-1
        new Thread(() -> flux.subscribe(i -> log.info(i))).start();

        TimeUnit.SECONDS.sleep(2);
    }

    /**
     * 4.6. 错误处理
     * 在反应式流中，错误是终端事件。一旦发生错误，它就会停止 序列，并沿着运算符链传播到最后一步，即 You 定义的及其方法。SubscriberonError
     */
    @Test
    public void errorHandling() {
        Flux.just(1, 2, 0)
                .map(i -> "100 / " + i + " = " + (100 / i)) //this triggers an error with 0
                .onErrorReturn("Divided by zero :(") // error handling example
                .subscribe(s -> log.info(s), e -> log.error("Error : {}", e));
    }

    /**
     * 4.6.1. 错误处理运算符
     * 您可能熟悉在 try-catch 块中处理异常的几种方法。 最值得注意的是，这些包括以下内容：
     * 捕获并返回静态默认值。
     * 使用回退方法捕获并执行备用路径。
     * 捕获并动态计算回退值。
     * 接住，包裹成 ，然后重新投掷。BusinessException
     * 捕获、记录特定于错误的消息，然后重新引发。
     * 使用该块清理资源或 Java 7 “try-with-resource” 构造。finally
     */
    @Test
    public void errorHandlingOperators() {
        Flux.just(1, 2, 0)
                .map(i -> "100 / " + i + " = " + (100 / i)) //this triggers an error with 0
                // 在订阅时执行的操作
//                .doOnSubscribe(subscription -> log.info("[subscription] {}", subscription))
                /**
                 * 捕获并吞下错误
                 * 如果您甚至不想将异常替换为回退值，而是忽略它，并且 只传播到目前为止已经产生的元素，你想要的本质上是替换 带有信号的信号。
                 * https://projectreactor.io/docs/core/release/reference/#_catch_and_swallow_the_error
                 */
//                .onErrorComplete()
                /**
                 * 静态回退值
                 * 捕获并返回静态默认值
                 * https://projectreactor.io/docs/core/release/reference/#_static_fallback_value
                 */
//                .onErrorReturn("[onErrorReturn]Divided by zero :(")
//                .onErrorReturn(e -> e instanceof ArithmeticException,"[onErrorReturn] onErrorReturn(Predicate<? super Throwable> predicate, T fallbackValue)")
//                .onErrorReturn(e -> e instanceof NullPointerException,"[onErrorReturn] onErrorReturn(Predicate<? super Throwable> predicate, T fallbackValue)")
                /**
                 * Fallback 方法
                 * 如果您想要多个默认值，并且您有另一种（更安全）的方法 处理您的数据时，您可以使用 .这相当于 “使用回退方法捕获并执行替代路径”。onErrorResume
                 * https://projectreactor.io/docs/core/release/reference/#_fallback_method
                 */
//                .flatMap(s -> Mono.just(0)
//                        .map(i -> (100 / i) + "Mono.just(0)")
//                        .onErrorResume(e -> Mono.just("Mono.just(0) error")))
//                // 	使用回退方法捕获并执行备用路径。onErrorResume
//                .onErrorResume(e -> {
//                    log.info("{}", e.toString());
//                    return Mono.just("[onErrorResume]Divided by zero :(");
//                })
                /**
                 * 动态回退值
                 * 即使您没有其他（更安全）的数据处理方式，您也可能需要 从收到的异常中计算回退值。这将是 等效于“捕获并动态计算回退值”。
                 * https://projectreactor.io/docs/core/release/reference/#_dynamic_fallback_value
                 */
                // 	捕获并动态计算回退值。onErrorResume
//                .onErrorResume(e -> {
//                    log.info("{}", e.toString());
//                    if (e instanceof ArithmeticException) {
//                        return Mono.just("[ArithmeticException]Divided by zero :(");
//                    } else if (e instanceof NullPointerException) {
//                        return Mono.just("[NullPointerException]NPE :(");
//                    }
//                    return Mono.just("Unknown :(");
//                })
                /**
                 * 接住并重新投掷
                 * 如果您想要在捕获异常后执行一些操作，但仍然希望 重新引发它，您可以使用 .这相当于 “接住，包裹成 ，然后重新投掷”。
                 * https://projectreactor.io/docs/core/release/reference/#_catch_and_rethrow
                 */
                // 	接住，包裹成 ，然后重新投掷。BusinessException
                .onErrorResume(original -> {
                    log.info("{}", original.toString());
                    return Flux.error(new RuntimeException("SLA exceeded", original));
                })
                // 	捕获、记录特定于错误的消息，然后重新引发。onErrorMap
                .onErrorMap(e -> {
                    log.info("{}", e.toString());
                    return new IllegalStateException("SLA exceeded", e);
                })
                /**
                 * 在一旁记录或反应
                 * 对于希望错误继续传播但仍希望做出反应的情况 在不修改序列（例如记录序列）的情况下，您可以使用运算符。这相当于“捕获、记录特定于错误的消息并重新引发” 模式
                 * https://projectreactor.io/docs/core/release/reference/#_log_or_react_on_the_side
                 */
                .doOnError(e -> log.info("[doOnError] : {}", e.toString()))
                /**
                 * 使用资源和 finally 块
                 * 与命令式编程的最后一个相似之处是可以完成的清理 通过使用“使用块清理资源”或使用 “Java 7 try-with-resource 构造”
                 * https://projectreactor.io/docs/core/release/reference/#_using_resources_and_the_finally_block
                 */
                // 	使用该块清理资源或 Java 7 “try-with-resource” 构造。finally
                .doFinally(type -> log.info("[doFinally] : {}", type))
                // 从上游请求正好 1，并在发出一个项目后取消
//                .take(1)
                .subscribe(s -> log.info(s), e -> log.error("Error : {}", e.toString()));
    }

    /**
     * 使用资源和 finally 块
     * 与命令式编程的最后一个相似之处是可以完成的清理 通过使用“使用块清理资源”或使用 “Java 7 try-with-resource 构造”，如下所示：finally
     * <p>
     * 例 14.命令式使用 finally
     * Stats stats = new Stats();
     * stats.startTimer();
     * try {
     * doSomethingDangerous();
     * }
     * finally {
     * stats.stopTimerAndRecordTiming();
     * }
     * <p>
     * 例 15.命令性地使用 try-with-resource
     * try (SomeAutoCloseable disposableInstance = new SomeAutoCloseable()) {
     * return disposableInstance.toString();
     * }
     */
    @Test
    public void doFinally() {
        LongAdder statsCancel = new LongAdder();
        Flux<String> flux =
                Flux.just("foo", "bar")
                        .doOnSubscribe(s -> log.info("subscribe : {}", s))
                        .doFinally(type -> {
                            log.info("type : {}", type);
                            if (type == SignalType.CANCEL)
                                statsCancel.increment();
                        })
                        .take(1);
        flux.subscribe(s -> log.info("subscribe : {}", s));
    }

    /**
     * 例 16.一次性资源
     * 现在我们可以对它执行“try-with-resource”的反应式等价物
     * 例 17.反应式 try-with-resource：using()
     */
    @Test
    public void using() {

        AtomicBoolean isDisposed = new AtomicBoolean();
        Disposable disposableInstance = new Disposable() {
            @Override
            public void dispose() {
                log.info("dispose");
                isDisposed.set(true);
            }

            @Override
            public String toString() {
                return "DISPOSABLE";
            }
        };

        Flux<String> flux = Flux.using(
                () -> disposableInstance,
                disposable -> Flux.just(disposable.toString()),
                Disposable::dispose
        );
        flux.subscribe(s -> log.info("subscribe : {}", s));

    }

    /**
     * 演示终端方面onError(interval)
     * 为了证明所有这些算子都会导致上游原始序列 terminate 当发生错误时，我们可以使用一个更直观的例子。运算符每 x 个时间单位勾选一次，其值递增。
     * 即使多了一秒钟的运行时间，也不会再有 滴答声传入。这 序列确实因错误而终止。interval
     * https://projectreactor.io/docs/core/release/reference/#_demonstrating_the_terminal_aspect_of_onerror
     */
    @Test
    public void interval() throws InterruptedException {
        Flux<String> flux =
                Flux.interval(Duration.ofMillis(250))
                        .map(input -> {
                            if (input < 3) return "tick " + input;
                            throw new RuntimeException("boom");
                        })
                        .onErrorReturn("Uh oh");

        flux.subscribe(s -> log.info("subscribe : {}", s), e -> log.error("Error : {}", e.toString()));
        TimeUnit.SECONDS.sleep(2);
    }

    /**
     * 重试 (异步)
     * 在错误处理方面，还有另一个感兴趣的运算符，您可能是 很想在上一节所述的情况下使用它。，正如它的名字 指示，允许您重试产生错误的序列。retry
     * https://projectreactor.io/docs/core/release/reference/#_retrying
     * <p>
     * subscribe : [268,tick 0]
     * subscribe : [248,tick 1]
     * subscribe : [246,tick 2]
     * 一个新的开始，从刻度 0 开始。额外的 250 毫秒持续时间为 来自第 4 个分时，导致异常的分时和随后的分时 重试
     * subscribe : [510,tick 0]
     * subscribe : [248,tick 1]
     * subscribe : [248,tick 2]
     * Error : java.lang.RuntimeException: boom
     */
    @Test
    public void retry() throws InterruptedException {
        Flux<String> flux =
                Flux.interval(Duration.ofMillis(250))
                        .map(input -> {
                            if (input < 3) return "tick " + input;
                            throw new RuntimeException("boom");
                        })
                        // 	重试产生错误的序列。retry
                        .retry(1);
        // elapsed 返回一个 Flux，它发出一个元素，该元素是相对于前一个元素的持续时间。elapsed
        flux.elapsed().subscribe(s -> log.info("subscribe : {}", s), e -> log.error("Error : {}", e.toString()));
        TimeUnit.SECONDS.sleep(5);
    }

    /**
     * 重试周期如下： (同步)
     * 每次发生错误（有可能重试）时，都会向 伴侣，已由您的功能装饰。在这里有一个 鸟瞰迄今为止的所有尝试。提供对错误的访问权限 以及围绕它的元数据。RetrySignalFluxFluxRetrySignal
     * 如果随播对象发出值，则会重试。Flux
     * 如果伴随完成，则吞下错误，重试周期停止， 生成的序列也完成。Flux
     * 如果伴随事件产生错误 （），则重试周期将停止，并且 导致的序列错误。Fluxee
     */
    @Test
    public void retryWhen() {
        Flux<String> flux = Flux
                // 这会不断产生错误，需要重试。
                .<String>error(new IllegalArgumentException())
                // doOnError在重试之前，我们可以记录并查看所有失败。
                .doOnError(e -> log.info("error : {}", e.toString()))
                // 它改编自一个非常简单的 lambda RetryFunction
                .retryWhen(Retry.from(companion ->
                        // 在这里，我们将前三个错误视为可重试 （），然后放弃。take(3)
                        companion.take(3)));
        flux.subscribe(s -> log.info("subscribe : {}", s), e -> log.error("Error : {}", e.toString()));
    }

    /**
     * 4.6.2. 处理运算符或函数中的异常、
     * 通常，所有运算符本身都可以包含可能触发 异常或对用户定义的回调的调用，这些回调同样可能失败，因此它们都可能失败 包含某种形式的错误处理
     * https://projectreactor.io/docs/core/release/reference/#_handling_exceptions_in_operators_or_functions
     */
    @Test
    public void handlingExceptionsInOperatorsOrFunctions() {
        mapThrow();
        log.info("--------------------------------------------------");
        converted();
    }

    public void converted() {

        Flux<String> converted = Flux
                .range(1, 10)
                .map(i -> {
                    try {
                        return convert(i);
                    } catch (IOException e) {
                        throw Exceptions.propagate(e);
                    }
                });

        converted.subscribe(
                v -> log.info("RECEIVED: " + v),
                e -> {
                    if (Exceptions.unwrap(e) instanceof IOException) {
                        log.info("Something bad happened with I/O");
                    } else {
                        log.info("Something bad happened");
                    }
                }
        );
    }

    public String convert(int i) throws IOException {
        if (i > 3) {
            throw new IOException("boom " + i);
        }
        return "OK " + i;
    }

    public void mapThrow() {
        Flux.just("foo")
                .map(s -> {
                    throw new IllegalArgumentException(s);
                })
                .subscribe(v -> log.info("GOT VALUE"),
                        e -> log.info("ERROR: " + e));
    }

    /**
     * 4.7. 接收器
     * 在 Reactor 中，接收器是一个允许以独立方式安全手动触发信号的类，创建一个能够处理多个（口味除外）的类似结构。Publisher Subscriber unicast()
     * <p>
     * 这些类别是：Sinks
     * many().multicast()：一个接收器，它将仅将新推送的数据传输给其订阅者，并尊重他们的背压（新推送的，如“订阅者订阅后”）。
     * many().unicast()：与上述相同，但对第一个订阅者注册之前推送的数据进行了缓冲。
     * many().replay()：一个接收器，它将向新订阅者重播推送数据的指定历史大小，然后继续实时推送新数据。
     * one()：将向其订阅者播放单个元素的接收器
     * empty()：一个接收器，它只向其订阅者播放终端信号（错误或完整），但仍然可以被视为（注意泛型类型）。Mono<T><T>
     */
    @Test
    public void sinks() {
        Sinks.Many<Integer> replaySink = Sinks.many().replay().all();
        //thread1
        replaySink.emitNext(1, Sinks.EmitFailureHandler.FAIL_FAST);

        //thread2,后来
        replaySink.emitNext(2, Sinks.EmitFailureHandler.FAIL_FAST);

        //线程3与线程2并发
        //将重试发射2秒，如果不成功则以EmissionException失败
        replaySink.emitNext(3, Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(2)));

        //线程3与线程2并发
        //返回FAIL_NON_SERIALIZED
        Sinks.EmitResult result = replaySink.tryEmitNext(4);
        log.info("{}", result);

        Flux<Integer> fluxView = replaySink.asFlux();
        fluxView
                .takeWhile(i -> i < 10)
                .log()
                .blockLast();
    }

    /**
     * 9.1. 共同使用算子
     */

    /**
     * 9.1.1. 使用运算符transform
     * 运算符允许您将运算符链的一部分封装到 功能。该函数在组装时应用于原始算子链，以 使用封装的运算符对其进行扩充。这样做会将相同的操作应用于所有 序列的订阅者，基本上等同于链接运算符 径直。
     * https://projectreactor.io/docs/core/release/reference/#_using_the_transform_operator
     */
    @Test
    public void transform() {
        Function<Flux<String>, Flux<String>> filterAndMap =
                f -> f.filter(color -> !color.equals("orange"))
                        .map(String::toUpperCase);

        Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                .doOnNext(s -> log.info("=== {}", s))
                .transform(filterAndMap)
                .subscribe(d -> log.info("Subscriber to Transformed MapAndFilter: {}", d));
    }

    /**
     * 9.1.2. 使用运算符transformDeferred
     * 运算符类似于，也允许您封装运算符 在函数中。主要区别在于此功能应用于原始 按每个订阅者排序。这意味着该函数实际上可以产生一个 每个订阅都有不同的运算符链（通过维护某种状态）。这 示例代码如下：transformDeferred transform
     */
    @Test
    public void transformDeferred() {
        AtomicInteger ai = new AtomicInteger();
        Function<Flux<String>, Flux<String>> filterAndMap = f -> {
            if (ai.incrementAndGet() == 1) {
                return f.filter(color -> !color.equals("orange"))
                        .map(String::toUpperCase);
            }
            return f.filter(color -> !color.equals("purple"))
                    .map(String::toUpperCase);
        };

        Flux<String> composedFlux =
                Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                        .doOnNext(s -> log.info("=== {}", s))
                        .transformDeferred(filterAndMap);

        composedFlux.subscribe(d -> log.info("Subscriber 1 to Composed MapAndFilter :" + d));
        composedFlux.subscribe(d -> log.info("Subscriber 2 to Composed MapAndFilter: " + d));
    }

    /**
     * 在 Reactor 中，transformDeferred 和 transform 是用于对流进行转换的方法。它们之间的区别在于懒加载和即时加载：
     *
     * transformDeferred:
     *
     * transformDeferred 方法会延迟执行转换逻辑。具体来说，它会在每次有订阅者（subscriber）时才调用转换函数。
     * 这意味着转换逻辑将在每个新的订阅者上下文中执行，并且可以根据每个订阅者的需求动态生成转换逻辑。
     *
     * transform:
     *
     * 相比之下，transform 方法会立即执行转换逻辑。一旦您调用 transform 并提供了转换函数，该函数就会立即应用于流。
     * 这意味着转换逻辑无论是否有订阅者都会被执行，并且在整个流的生命周期内都使用相同的转换逻辑。
     * 因此，主要区别在于 transformDeferred 是一种懒加载的方式，它允许在每个订阅者上下文中动态执行转换逻辑，而 transform 则是一种即时加载的方式，它会立即应用转换逻辑到整个流上。
     */


    /**
     * 9.3. 向多个订阅者广播ConnectableFlux
     * 有时，您可能不希望仅将某些处理推迟到一个订阅时间 订阅者，但您实际上可能希望其中几个会合，然后 触发订阅和数据生成。
     * 这就是它的用途。API 中涵盖了两种主要模式，它们返回 ： 和 。ConnectableFluxFluxConnectableFluxpublishreplay
     * publish动态地尝试尊重其各个订阅者的需求，在 通过将这些请求转发给源。最值得注意的是，如果有的话 subscriber 有一个待处理的请求 ，publish 暂停其对源的请求。0
     * replay缓冲通过第一个订阅看到的数据，直至可配置的限制 （在时间和缓冲区大小上）。它将数据重播给后续订阅者。
     * A 提供了管理下游订阅的其他方法 与订阅原始来源相比。这些附加方法包括 以后：ConnectableFlux
     * connect()一旦达到足够的订阅，就可以手动调用。那 触发对上游源的订阅。Flux
     * autoConnect(n)订阅后可以自动执行相同的工作 䍬。n
     * refCount(n)不仅可以自动跟踪传入的订阅，还可以检测 当这些订阅被取消时。如果跟踪的订阅者不足，则来源 是“断开连接”，如果其他 此时将显示订阅者。
     * refCount(int, Duration)添加“宽限期”。一旦跟踪的订阅者数量 变得太低，它会等待 在断开源之前，可能 允许足够多的新订阅者进入并再次跨越连接阈值。Duration
     */
    @Test
    public void connect() throws InterruptedException {

        Flux<Integer> source = Flux.range(1, 3)
                .doOnSubscribe(s -> log.info("subscribed to source"));

        ConnectableFlux<Integer> co = source.publish();

        co.subscribe(i -> log.info("one : {}", i), e -> log.error("one : {}", e), () -> {
        });
        co.subscribe(i -> log.info("two : {}", i), e -> log.error("two : {}", e), () -> {
        });

        log.info("done subscribing");
        TimeUnit.SECONDS.sleep(5);
        log.info("will now connect");

        co.connect();

    }

    @Test
    public void autoConnect() throws InterruptedException {

        Flux<Integer> source = Flux.range(1, 3)
                .doOnSubscribe(s -> log.info("subscribed to source"));

        Flux<Integer> autoCo = source.publish().autoConnect(2);

        autoCo.subscribe(i -> log.info("one : {}", i), e -> log.error("one : {}", e), () -> {
        });
        log.info("subscribed first");
        TimeUnit.SECONDS.sleep(5);
        log.info("subscribing second");
        autoCo.subscribe(i -> log.info("two : {}", i), e -> log.error("two : {}", e), () -> {
        });
    }

    /**
     * 9.4.1. 分组Flux<GroupedFlux<T>>
     * 分组是将源拆分为多个批次的行为，每个批次 匹配密钥。Flux<T>
     * 关联的运算符是 。groupBy
     * 每个组都表示为 ，它允许您通过调用其方法来检索键。GroupedFlux<T>key()
     * 小组的内容没有必要的连续性。曾经是源元素 生成一个新键，此键的组将打开，并且与键匹配的元素结束 在组中（可以同时打开多个组）。
     * 这意味着组：
     * 始终不相交（源元素属于一个且仅属于一个组）。
     * 可以包含来自原始序列中不同位置的元素。
     * 从不为空。
     */
    @Test
    public void groupBy() {
        Flux.range(1, 10)
                .groupBy(i -> i % 2 == 0 ? "even" : "odd")
                .doOnNext(g -> log.info("group : {}, body : {}", g.key(), g))
                // 所以concatMap的作用就是对每一个分组的Flux执行，并将最后的结果连接在一起。
                // concatMap保证了原始数据的顺序，也就是说，只有当前面一组的所有元素都处理完毕后，才会处理下一组。
                .concatMap(g -> g.defaultIfEmpty(-1) //如果是空组，则显示它们
                        .map(String::valueOf) //映射到字符串
                        .startWith(g.key())) //从组键开始
                .subscribe(s -> log.info("subscribe : {}", s));
    }

    /**
     * 9.5. 并行化工作ParallelFlux
     * 随着多核架构成为当今的商品，能够轻松 并行化工作很重要。Reactor 通过提供一种特殊类型来帮助实现这一点，该类型公开了针对并行化工作进行优化的运算符。ParallelFlux
     * 要获得 ，可以在任何 上使用运算符。 就其本身而言，此方法不会并行化工作。相反，它分裂了 将工作负载放入“Rail”（默认情况下，有多少个 CPU 内核，就有多少个 Rail）。ParallelFluxparallel()Flux
     * 为了告诉结果在哪里运行每条轨道（并且，通过 扩展，以并行运行 rails），您必须使用 .请注意， 建议将并行工作专用于：。ParallelFluxrunOn(Scheduler)SchedulerSchedulers.parallel()
     */
    @Test
    public void parallelFlux(){

        Flux.range(1, 10)
                .parallel(2)
                .subscribe(i -> log.info(Thread.currentThread().getName() + " -> " + i));

        log.info("==========================================================");

        Flux.range(1, 10)
                .parallel(2)
                // runOn 会将每个轨道的工作分发到不同的线程上
                .runOn(Schedulers.parallel())
                .subscribe(i -> log.info(Thread.currentThread().getName() + " -> " + i));

    }

    static final String HTTP_CORRELATION_ID = "reactive.http.library.correlationId";

    @Test
    public void http() {

        Mono<Tuple2<Integer, String>> mono = doPut("www.example.com", Mono.just("Walter"));
        log.info("{}", mono.block());

        Mono<Tuple2<Integer, String>> mono1 = doPut("www.example.com", Mono.just("Walter"))
                .contextWrite(Context.of(HTTP_CORRELATION_ID, "2-j3r9afaf92j-afkaf"));
        log.info("{}", mono1.block());
    }

    Mono<Tuple2<Integer, String>> doPut(String url, Mono<String> data) {
        Mono<Tuple2<String, Optional<Object>>> dataAndContext =
                data.zipWith(Mono.deferContextual(c ->
                        Mono.just(c.getOrEmpty(HTTP_CORRELATION_ID)))
                );

        return dataAndContext.<String>handle((dac, sink) -> {
                    if (dac.getT2().isPresent()) {
                        sink.next("PUT <" + dac.getT1() + "> sent to " + url +
                                " with header X-Correlation-ID = " + dac.getT2().get());
                    }
                    else {
                        sink.next("PUT <" + dac.getT1() + "> sent to " + url);
                    }
                    sink.complete();
                })
                .map(msg -> Tuples.of(200, msg));
    }

}
